package com.heima.search.listener;

import com.heima.common.redis.RedisCacheService;
import com.heima.model.common.constants.ArticleConstants;
import com.heima.model.search.dtos.SearchArticleVo;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

import static com.heima.common.constants.DelayConstants.DELAY_WNMATERIAL_KEY;

@Slf4j
@Component
public class SyncArticleListener {

    @Autowired
    private RestHighLevelClient esClient; // 建议重命名为esClient更清晰
    @Autowired
    private RedisCacheService redisCacheService;


    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(name = ArticleConstants.ARTICLE_ES_SYNC_QUEUE, durable = "true"),
                    exchange = @Exchange(name = ArticleConstants.ARTICLE_ES_SYNC_TOPIC, type = ExchangeTypes.TOPIC),
                    key = ArticleConstants.ARTICLE_ES_SYNC_ROUTINGKEY
            )
    )
    public void syncArticleToEs(SearchArticleVo searchArticleVo) {
        // ========== 1. 参数校验 ==========
        if (searchArticleVo == null || searchArticleVo.getId() == null) {
            log.error("接收无效消息：SearchArticleVo为空或id为null → {}", searchArticleVo);
            return;
        }
        Long articleId = searchArticleVo.getId();
        String lockKey = DELAY_WNMATERIAL_KEY + ":" + articleId + ":" + System.currentTimeMillis() / 1000;
        String lockToken = null;


        try {
            // ========== 2. 获取Redis分布式锁（避免重复同步） ==========
            lockToken = redisCacheService.tryLock(lockKey, 60); // 锁有效期60秒
            if (lockToken == null) {
                log.warn("文章{}已被其他线程同步，跳过本次处理", articleId);
                return;
            }


            // ========== 3. 构建ES更新请求 ==========
            UpdateRequest updateRequest = new UpdateRequest(
                    "app_info_article",  // 索引名
                    articleId.toString() // 文档ID
            );

            // 将SearchArticleVo转换为ES文档（支持新增或更新）
            Map<String, Object> docMap = new HashMap<>();
            docMap.put("title", searchArticleVo.getTitle());
            docMap.put("content", searchArticleVo.getContent());
            docMap.put("publishTime", searchArticleVo.getPublishTime());
            docMap.put("authorId", searchArticleVo.getAuthorId());
            docMap.put("authorName", searchArticleVo.getAuthorName());
            docMap.put("images", searchArticleVo.getImages());
            docMap.put("layout", searchArticleVo.getLayout());
            docMap.put("staticUrl", searchArticleVo.getStaticUrl());

            // 设置更新内容（如果文档不存在，则自动插入docMap）
            updateRequest.doc(docMap, XContentType.JSON)
                    .upsert(docMap, XContentType.JSON); // upsert：不存在则插入


            // ========== 4. 执行ES更新 ==========
            UpdateResponse response = esClient.update(updateRequest, RequestOptions.DEFAULT);
            log.info("文章{}同步到ES成功，结果：{}", articleId, response.getResult());


        } catch (Exception e) { // 捕获具体异常（如IOException）
            log.error("文章{}同步到ES失败：{}", articleId, e.getMessage(), e);
            throw new RuntimeException("ES文章同步失败", e); // 抛出带原因的异常
        } finally {
            // ========== 5. 释放Redis锁 ==========
            if (lockToken != null) {
                redisCacheService.unlock(lockKey, lockToken);
                log.info("释放文章{}的同步锁", articleId);
            }
        }
    }
}